package org.aurora.remote.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.aurora.common.serialization.Byte2ObjectSerializationFactory;
import org.aurora.common.util.OSUtil;
import org.aurora.remote.RemoteClient;
import org.aurora.remote.RemoteProcessor;
import org.aurora.remote.SendCallback;
import org.aurora.remote.protocol.RemoteRequest;
import org.aurora.remote.protocol.RemoteResponse;

import java.net.InetSocketAddress;

public class NettyRemoteClient implements RemoteClient<NettyEventSubscriber> {
    private String serverIp;
    private int serverPort;

    private Bootstrap bootstrap;
    private Channel channel;
    private EventLoopGroup eventLoopGroup;

    private NettyRemoteCommandHandler nettyRemoteCommandHandler;

    public NettyRemoteClient(String serverIp, int serverPort) {
        this.serverPort = serverPort;
        this.serverIp = serverIp;
        this.nettyRemoteCommandHandler = new NettyRemoteCommandHandler();

        boolean isLinux = OSUtil.isLinux();

        this.eventLoopGroup = isLinux ? new EpollEventLoopGroup() : new NioEventLoopGroup(1); // todo 需不需要设置为多个线程
        this.bootstrap = new Bootstrap();
        this.bootstrap
                .group(eventLoopGroup)
                .channel(isLinux ? EpollSocketChannel.class : NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(NettyRemoteClient.this.nettyRemoteCommandHandler.getNettyConnectChangeHandler())
                                .addLast(NettyEncoder.getInstance())
                                .addLast(new NettyDecoder())
                                .addLast(nettyRemoteCommandHandler);
                    }
                });
    }

    @Override
    public void start() throws InterruptedException {
        ChannelFuture channelFuture = this.bootstrap.connect(serverIp, serverPort).sync();
        this.channel = channelFuture.channel();
        this.nettyRemoteCommandHandler.start();
    }

    @Override
    public void shutdown() {
        if (this.channel != null) {
            this.channel.close();
        }
    }

    @Override
    public void registerProcessor(int commandCode, RemoteProcessor<RemoteRequest, RemoteResponse> processor) {
        this.nettyRemoteCommandHandler.registerProcessor(commandCode, processor);
    }

    @Override
    public void registerEventManager(NettyEventSubscriber remoteConnectionEventManager) {
        this.nettyRemoteCommandHandler.registerEventSubscriber(remoteConnectionEventManager);
    }

    @Override
    public <RESPONSE extends RemoteResponse> RESPONSE send(RemoteRequest request, long timeoutMillis, Class<RESPONSE> responseClazz) {
        RemoteResponse remoteResponse = nettyRemoteCommandHandler.send(channel, request, timeoutMillis);

        return Byte2ObjectSerializationFactory.deserialization(remoteResponse.getData(), responseClazz, remoteResponse.getSerializationType());
    }

    @Override
    public <RESPONSE extends RemoteResponse> void sendAsync(RemoteRequest request, long timeoutMillis, SendCallback<RESPONSE> sendCallback) {
        nettyRemoteCommandHandler.sendAsync(channel, request, timeoutMillis, sendCallback);
    }

    @Override
    public String getHost() {
        return ((InetSocketAddress)channel.localAddress()).getHostString();
    }

    @Override
    public int getPort() {
        return ((InetSocketAddress)channel.localAddress()).getPort();
    }
}
